package cn.xinfei.xdecision.engine.runner.producer;

import cn.xinfei.xdecision.nacos.NacosSwitchProperties;
import cn.xinfei.xdecision.pulsar.XPulsarProducer;
import cn.xinfei.xdecision.pulsar.config.XDecisionMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Map;
import java.util.concurrent.CompletableFuture;

@Slf4j
public class AbsDecisionProducer {

    @Autowired
    private XPulsarProducer xPulsarProducer;


    public final boolean send(String topic, XDecisionMessage msg, int retry) throws Exception {
        long start = System.currentTimeMillis();
        try {
            CompletableFuture<MessageId> completableFuture = xPulsarProducer.sendAsync(topic, msg);
            log.info("message sending,topic={},msgId={},retry={}", topic, msg.getMsgId(), retry);
//            MessageId messageId = completableFuture.get(10, TimeUnit.SECONDS);
            MessageId messageId = completableFuture.get();
            long end = System.currentTimeMillis();
            log.info("message sended,topic={},msgId={},messageId={},producerDuration={}", topic, msg.getMsgId(), messageId.toString(), end - start);
        } catch (Exception e) {
            //此处进行了重试，consumer如果是update的场景如何进行幂等？现在update没有做幂等...
            log.error("message send fail,进行重试,topic={},msgId={}", topic, msg.getMsgId(), e);
            if (retry < 3) {
                send(topic, msg, ++retry);
            } else {
                log.error("message send fail,不再重试,topic={},msgId={}", topic, msg.getMsgId(), e);
            }
        }

        return true;
    }

    public  void sendAsync(String topic, XDecisionMessage msg,int retry) throws PulsarClientException {
        long start = System.currentTimeMillis();
        CompletableFuture<MessageId> completableFuture = xPulsarProducer.sendAsync(topic, msg);
        Map<String, String> copyOfContextMap = MDC.getCopyOfContextMap();
        long ending = System.currentTimeMillis();

        if (NacosSwitchProperties.getNacosSwitchProperties().isDebugLog()) {
            log.info("message sending,topic={},msgId={},sendDuration={},retry={}", topic, msg.getMsgId(), ending - start, retry);
        }
        completableFuture.handleAsync((msgId,ex) -> {
            MDC.setContextMap(copyOfContextMap);
            if(ex==null) {
                long end = System.currentTimeMillis();
                log.info("message send success,topic={},msgId={},messageId={},producerDuration={},retry={}", topic, msg.getMsgId(), msgId.toString(), end - start, retry);
            }else{
                long end = System.currentTimeMillis();
                log.error("message send fail,topic={},msgId={},producerDuration={},retry={}", topic, msg.getMsgId(), end - start,retry,ex);
            }

//            if (finalRetry < 3) {
//                try {
//                    sendAsync(topic, msg, ++finalRetry);
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
//            } else {
//                log.error("message send fail,不再重试,topic={},msgId={},retry={}", topic, msg.getMsgId(),retry, ex);
//            }
            return null;
        });
    }
}
